package org.nbict.iot.trash;

import com.alibaba.fastjson.JSONObject;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.nbict.iot.persist.pool.DbRepository;
import org.nbict.iot.persist.repo.MongoRepo;

import java.util.Map;

/**
 * Created by songseven on 18/6/20.
 */
@Deprecated
public class DispatcherRealPacketBolt extends BaseBasicBolt {

    private MongoRepo repo;

    public void prepare(Map stormConf, TopologyContext context) {
        repo = (MongoRepo) DbRepository.MONGO.getInstance();
    }

    @Override
    public void execute(Tuple input, BasicOutputCollector
            collector) {

        String parsed = (String) input.getValueByField("value");
        JSONObject object = JSONObject.parseObject(parsed);
        JSONObject body = (JSONObject) object.getJSONObject("car_packet_body");
        JSONObject header = (JSONObject) object.getJSONObject
                ("car_packet_header");

        String platform = (String) input.getValueByField("topic");
        String vin = header.getString("car_vin");
        int cmd = header.getInteger("header_cmd");
        Long timestamp = body.getLong("timestamp");

        System.out.println("tag ==> " + cmd);

        if (cmd == 2 || cmd == 3) { //实时报文和补发报文解析
            JSONObject upload = body.getJSONObject("upload");
            for (String tag : upload.keySet()) {

                JSONObject info = (JSONObject) upload.get(tag);

                collector.emit("carRealtime", new Values(platform, vin, tag,
                        timestamp, info));

                if (tag.equals("01")) { //更新车辆状态
                    collector.emit("carInfoMap", new Values(vin, timestamp,
                            info.get("status"), info.get("charged_state")));
                } else if (tag.equals("05")) { //更新地理信息
                    collector.emit("carGeoMap", new Values(vin, timestamp,
                            info.get("lon"), info.get("lat")));
                } else if (tag.equals("07")) { //更新报警
                    collector.emit("carAlarmMap", new Values(vin,
                            timestamp, info.get("alarm_level")));
                }
            }
        } else if (cmd == 1) { //车辆登入

        } else if (cmd == 4) { //车辆登出

        } else if (cmd == 5 || cmd == 6) { //平台登入登出

        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream("carRealtime", new Fields("platform_id",
                "car_vin", "tag", "timestamp", "info")); //车辆实时流
        declarer.declareStream("carInfoMap", new Fields("car_vin",
                "timestamp", "car_status", "charge_status"));//地图整车流
        declarer.declareStream("carGeoMap", new Fields("car_vin",
                "timestamp", "lon", "lat")); //地理信息流
        declarer.declareStream("carAlarmMap", new Fields("car_vin",
                "timestamp", "alarm_status")); //告警流

    }
}
